package top.jolyoulu.streaming

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.util.Random

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_State {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.checkpoint("cp") //设置检查点，updateStateByKey需要使用检查点保留状态
    val datas: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.88.100", 9999)
    val wordToOne: DStream[(String, Int)] = datas.map((_, 1))
    //updateStateByKey 根据Key对数据状态进行跟新
    //参数1：相同Key的Value数据
    //参数2：缓冲区中相同Key的Value数据
    val state: DStream[(String, Int)] = wordToOne.updateStateByKey(
      (seq: Seq[Int], buff: Option[Int]) => {
        val newCount = buff.getOrElse(0) + seq.sum
        Option(newCount)
      }
    )
    state.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
